package com.danbay.framework.protocol.mqtt;

import com.danbay.framework.protocol.command.AbstractCommandHandler;
import com.danbay.framework.protocol.command.ICommandSelector;
import com.danbay.framework.protocol.command.ProtocolMessage;
import com.danbay.framework.protocol.command.codec.ICmdDataDecoder;
import com.danbay.framework.protocol.command.codec.ICmdDataEncoder;
import com.danbay.framework.protocol.command.codec.JsonDataDecoder;
import com.danbay.framework.protocol.command.codec.JsonDataEncoder;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 发布消息的回调类
 *
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
 * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
 * 在回调中，将它用来标识已经启动了该回调的哪个实例。
 * 必须在回调类中实现三个方法：
 *
 *  public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
 *
 *  public void connectionLost(Throwable cause)在断开连接时调用。
 *
 *  public void deliveryComplete(MqttDeliveryToken token))
 *  接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
 *  由 MqttClient.connect 激活此回调。
 *
 * @author SJG
 */
public class MqttCallback extends AbstractCommandHandler implements org.eclipse.paho.client.mqttv3.MqttCallback {

    private final ICommandSelector commandSelector;
    private final MqttClient mqttClient;

    @Autowired
    public MqttCallback(ICommandSelector commandSelector, MqttClient mqttClient) {
        this.commandSelector = commandSelector;
        this.mqttClient = mqttClient;
    }

    @Override
    public void connectionLost(Throwable e) {
        // 连接丢失后，一般在这里面进行重连
        logger.warn("connection lost,reconnect...",e);
        init();
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            MqttMessage mqttMessage=token.getMessage();
            String msg=mqttMessage==null?"":new String(mqttMessage.getPayload());
            logger.info("deliveryComplete,{},msg:{}",token.isComplete(),msg );
        } catch (MqttException e) {
            logger.error("get delivery msg error",e);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String msg=new String(message.getPayload());
        logger.info("id:{},topic:{},qos:{},payload:{}",message.getId(),topic,message.getQos(),msg);
        try {
            if(commandSelector!=null) {
                invokeCmd(new ProtocolMessage(topic, msg), commandSelector, getDataDecoder(), getDataEncoder());
            }
        }catch (Exception e){
            logger.error("invokeCmd error!",e);
        }
    }

    @Override
    public ICmdDataDecoder getDataDecoder() {
        return new JsonDataDecoder();
    }

    @Override
    public ICmdDataEncoder getDataEncoder() {
        return new JsonDataEncoder();
    }

    @Override
    public void init() {
        try {
            mqttClient.init(this);
        } catch (MqttException e) {
            logger.error("init mqtt client error!",e);
        }
    }
}
